package com.ww.hadoop.mr.fof;

import com.ww.hadoop.mr.topn.TKey;
import com.ww.hadoop.mr.topn.TMapper;
import lombok.SneakyThrows;
import lombok.extern.log4j.Log4j;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.util.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;

@Log4j
public class FMapper extends Mapper<LongWritable, Text, Text, IntWritable> {


    Text mkey = new Text();
    IntWritable mval = new IntWritable();


    @SneakyThrows
    @Override
    protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, IntWritable>.Context context) throws IOException, InterruptedException {
        //m y g z
        String[] s = value.toString().split(" ");
        for (int i = 1; i < s.length; i++) {
            String k = getFof(s[0], s[i]);
            mkey.set(k);
            mval.set(0);
            context.write(mkey, mval);
            for (int j = i + 1; j < s.length; j++) {
                String k1 = getFof(s[i], s[j]);
                mkey.set(k1);
                mval.set(1);
                context.write(mkey, mval);
            }
        }

    }

    public static String getFof(String s1, String s2) {

        if (s1.compareTo(s2) > 0) {
            return s1 + "-" + s2;
        } else {
            return s2 + "-" + s1;
        }

    }

}
